package com.atguigu.day02;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.phoenix.queryserver.client.ThinClientUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Random;

public class Flink06_Source_Custom_Phoenix {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.TODO 通过自定义的Source源获取数据
        DataStreamSource<WaterSensor> streamSource = env.addSource(new MySource()).setParallelism(1);

        streamSource.print();

        env.execute();
    }

    //    public static class MySource implements SourceFunction<WaterSensor>{
    //自定义Source从Phoenix读取数据
    public static class MySource implements ParallelSourceFunction<WaterSensor> {

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {

            String url = ThinClientUtil.getConnectionUrl("hadoop102", 8765);
            System.out.println(url);
            Connection connection = DriverManager.getConnection(url);

            //2.语句预执行
            PreparedStatement pstm = connection.prepareStatement("select * from student");

            //3.执行语句获取结果
            ResultSet resultSet = pstm.executeQuery();

            while (resultSet.next()) {
                System.out.println(resultSet.getString(1) + "-" + resultSet.getString(2) + "-" + resultSet.getString(3));
            }

            //4.关闭资源
            pstm.close();
            connection.close();

        }

        @Override
        public void cancel() {

        }
    }
}
